package htest;

import java.net.*;
import java.io.*;
import java.lang.*;
import java.lang.Math;
import java.lang.InterruptedException;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import org.w3c.dom.*;
import javax.xml.parsers.*;
import javax.xml.stream.*;

import org.apache.hadoop.fs.*;
import org.apache.hadoop.hdfs.*;
import org.apache.hadoop.hdfs.server.common.UpgradeStatusReport;
import org.apache.hadoop.hdfs.DistributedFileSystem.DiskStatus;
import org.apache.hadoop.hdfs.protocol.FSConstants.UpgradeAction;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.hdfs.protocol.*;
import org.apache.hadoop.conf.Configuration;

public class TestIO {

	private String shareDir;
	private String testCaseNo;
	private String globalPrefix;
	private int totalDataNode;
	private int totalClient;
	private String clientNo;
	private String stage;
	private String report;

	private int expired;
	private int round;

	private int writeThreads;
	private int fileSize;
	private long maxWriteBytes;

	private String readSource = null;
	private String writeCon = null;
	private int readThreads;
	private int step;
	private int readMode;

	private Timer schedule = null;

	private AtomicLong totalWriteByte     = new AtomicLong(0);
	private AtomicLong lastTotalWriteByte = new AtomicLong(0);
	private long totalReadByte      = 0;
	private AtomicLong lastTotalReadByte  = new AtomicLong(0);
	private AtomicInteger runTime         = new AtomicInteger(0);
	private AtomicInteger writeFiles      = new AtomicInteger(0);

	private Vector<String> totalFiles = new Vector<String>();

	FileWriter fileWriter = null;

	private long lastWriteByte = 0;
	private long lastReadByte  = 0;

	private Vector<WriteThread> writeThreadVector = new Vector<WriteThread>();
	private Vector<ReadThread> readThreadVector   = new Vector<ReadThread>();

	AtomicInteger write_running_threads = new AtomicInteger(0);
	AtomicInteger read_running_threads  = new AtomicInteger(0);

	private DistributedFileSystem myDFS = null;
	private Configuration myConf        = new Configuration();

	private class WriteThread extends Thread {
		// thread no
		int  threadNo;
		AtomicLong bytesWrite       = new AtomicLong(0);
		long lastBytesWrite         = 0;
		long blockCount             = 0;
		// We close this file, let blocked thread to throw IOException
		FSDataOutputStream currFout = null;

		public WriteThread(int threadNumber) {
			threadNo = threadNumber;
		}

		/**
		 * The main function of write thread
		 */
		public void run() {
			write_running_threads.incrementAndGet();
			System.out.println("WRITE: thread " + threadNo + " enter, running threads <--> " + write_running_threads.get());

			try {
				int bufferlen = myConf.getInt("io.file.buffer.size", 4096);
				byte buffer[] = new byte[bufferlen];
				for (int i = 0; i < bufferlen; i++)
					buffer[i] = Byte.parseByte(Integer.toString(i % 128));

				boolean is_loop = true;
				while (is_loop) {
					/* transfer ONE file */
					Path currWriteFile = new Path(build_hdfs_path(writeFiles.getAndIncrement()));
					if (myDFS.exists(currWriteFile)) {
						myDFS.delete(currWriteFile, false);
					}

					currFout = myDFS.create(currWriteFile);

					try {
						for (int i = 0; i < fileSize; i += bufferlen) {
							currFout.write(buffer, 0, bufferlen); 
							totalWriteByte.addAndGet(bufferlen);                                

							if (maxWriteBytes > 0 && totalWriteByte.get() >= maxWriteBytes) {
								is_loop = false;
								break;
							} else if (maxWriteBytes == 0 && runTime.get() >= expired) {
								is_loop = false;
								break;
							}
						}
					} catch (IOException e) {
						printErr(e.toString());
					} finally {
						try {
							currFout.close();
							currFout = null;
						} catch (IOException e) {
							printErr(e.toString());
						}
					}
				}
			} catch (IOException e) {
				printErr(e.toString());
			}

			write_running_threads.decrementAndGet();
			System.out.println("WRITE: thread [" + threadNo + "] exit, runnning threads <--> " + write_running_threads.get());
		}

		private void printErr(String str) {
			System.err.println("Thread [" + threadNo + "] Error: "+ str);
		}
	}

	private class ReadThread extends Thread {
		int threadNo;
		Vector<String> threadFileList = new Vector<String>();
        long threadReadByte=0;

		/**
		 * The main function of read thread.
		 * @precondition threadFileList is not empty
		 */
		public void run() {
			read_running_threads.incrementAndGet();
			System.out.println("READ: thread " + threadNo + " enter, running threads <--> " + read_running_threads.get());

			try {
                threadReadByte=0;
				int bufferlen = myConf.getInt("io.file.buffer.size", 4096);
				byte[] buffer = new byte[bufferlen];

				for(int i = 0; ; i++) {
					if (runTime.get() > expired)
						break;

					String fileName = threadFileList.get(i%threadFileList.size());

					Path readFile = new Path(fileName);
					if (!myDFS.exists(readFile)) {
						System.err.println("READ: No such file " + fileName);
						continue;
					}

					FSDataInputStream in = myDFS.open(readFile);

					System.out.println("READ: thread " + threadNo + " read file " + fileName);

					//long fileReadByte=0L;

					try {
						int bytesRead = 0;
						while (true) {
							bytesRead = in.read(buffer);
							if (bytesRead < 0)
								break;
                            threadReadByte+=bytesRead;
							//totalReadByte.addAndGet(bytesRead);
						}
					} catch (IOException e) {
						System.err.println(e.toString());
					} finally {
						in.close();
					}
				} 
			} catch (Exception e) {
				System.err.println(e.toString());
			}

			read_running_threads.decrementAndGet();
			System.out.println("READ: thread [" + threadNo + "] exit, running threads <--> " + read_running_threads.get());
		}

		public ReadThread(int threadNumber) throws IOException {
			threadNo = threadNumber;
			int tmp = -1;

			// XXX this can be ZERO!!!
			int filesPerThread = totalFiles.size() / readThreads;

			if (readMode < 0) { /** random read */
				System.out.println("Thread " + threadNo + ": random read " + filesPerThread + " files");

				Random r = new Random(); 
				for (int i = 0; i < filesPerThread; ++i) {
					threadFileList.add(totalFiles.get(Math.abs(r.nextInt() % totalFiles.size())));
				}
			} else if (0 == readMode) { /** sequential read */
				System.out.println("Thread " + threadNo + ": sequential read " + filesPerThread + " files");

				for (int i = 0; i < filesPerThread; i++) {
					threadFileList.add(totalFiles.get(threadNo *filesPerThread + i));
				}
			} else { /** reduplicate read */
				System.out.println("Thread " + threadNo + ": reduplicate read " + filesPerThread + " files");

				for (int i = 0; i < filesPerThread; ++i) {
					if ((i >= filesPerThread - filesPerThread*readMode/100) && (-1 == tmp)) 
						tmp = i;

					if (-1 == tmp)
						threadFileList.add(totalFiles.get(threadNo * filesPerThread + i));
					else
						threadFileList.add(totalFiles.get(threadNo * filesPerThread + tmp));
				}
			}

			write_thread_flist(threadFileList);
		}

		/**
		 * Write each thread read file list.
		 */
		private void write_thread_flist(Vector<String> flist) throws IOException {
			String path = build_local_path();
			path += "/stage_" + stage + "_thread_" + threadNo + ".filelist";

			System.out.println("write file list to " + path + " [ " + flist.size() + " items ]");

			StringBuffer buffer = new StringBuffer();
			for (int i = 0; i < flist.size(); i++) {
				buffer.append(flist.get(i) + "\n");
			}

			write_to_file(path, buffer.toString());
		}
	}

	public TestIO(String sdir, String prefix, String tno, int dnum, 
			int cnum, int fsize, int expire, int rthread, int wthread,
			int rno, String cno, String rept, int rmode, 
			int stp, String wcon, String stge, String rsrc, long totalbyte) {
		shareDir = sdir;
		globalPrefix = prefix;
		testCaseNo = tno;
		totalDataNode = dnum;
		totalClient = cnum;
		fileSize = fsize;
		expired = expire;
		readThreads = rthread;
		writeThreads = wthread;
		round = rno;
		clientNo = cno;
		report = rept;
		readMode = rmode;
		writeCon = wcon;
		step = stp;
		stage = stge;
		readSource = rsrc;
		maxWriteBytes = totalbyte;

		try {
			myDFS = (DistributedFileSystem) FileSystem.get(myConf);
		} catch (IOException e) {
			System.err.println(e.toString());
		}
	}

	private void check_threads() {

		for (WriteThread wt: writeThreadVector) {
			if (wt.lastBytesWrite != wt.bytesWrite.get()) {
				wt.lastBytesWrite = wt.bytesWrite.get();
				wt.blockCount = 0;
			}
			else
				wt.blockCount++;

			if (wt.blockCount > 60) {
				System.err.println("###### write thread " + wt.threadNo + " is blocked!!!");
				// cancel this thread, and then create a new thread
				try {
					if (wt.currFout != null) {
						wt.currFout.close();
						wt.currFout = null;
					}
				} catch (IOException e) {
					System.err.println(e.toString());
				}

				// reset
				wt.blockCount = 0;
			}
		}
	}

	private void write_filelist() throws IOException {
		String path = build_local_path();
		File f = new File(path);
		if (!f.exists()) 
			f.mkdirs();

		path += "/" + "stage_" + stage + ".filelist";

		System.out.println("write file list to " + path + " [ " + writeFiles.get() + " items ]");

		StringBuffer buffer = new StringBuffer();
		for (int i = 0; i < writeFiles.get(); i++) {
			buffer.append(build_hdfs_path(i) + "\n");
		}

		write_to_file(path, buffer.toString());
	}

	private void write_total_flist(Vector<String> flist) throws IOException {
		String path = build_local_path();
		path += "/stage_" + stage + "_total.filelist";

		System.out.println("write file list to " + path + " [ " + flist.size() + " items ]");

		StringBuffer buffer = new StringBuffer();
		for (int i = 0; i < flist.size(); i++) {
			buffer.append(flist.get(i) + "\n");
		}

		write_to_file(path, buffer.toString());
	}

	/** 
	 * Main function, we registed the scheduler which could wake up each 
	 * second and do some job for us, then we launch all read and write thread.
	 * This function has no argument and no return value.
	 */
	public void start() throws IOException {

		/**
		 * ONLY report
		 */
		if (report.equalsIgnoreCase("only_report")) {
			write_report();
			return;
		}

		if (schedule == null)
			schedule = new Timer();

		if (fileWriter == null)
			write_io_begin();

		final int interval = 20;
		//final int interval = 120;

		schedule.scheduleAtFixedRate( 
				new TimerTask() {
				public void run() {

                totalReadByte=0L;
				for (ReadThread t:readThreadVector) {
                    totalReadByte+=t.threadReadByte;
                }

				/** Start record only if the threads start read or write data*/
				if (totalWriteByte.longValue() == 0 && totalReadByte == 0L) {
				return;
				}
				runTime.addAndGet(interval);

				long localtime = System.currentTimeMillis();
				long writeByte = totalWriteByte.longValue();
				long readByte  = totalReadByte;

				System.out.println("*** Flying time: " + runTime.get() +
					" wb " + writeByte +
					" wt " + write_running_threads.get() + 
					" wr " + ((writeByte-lastWriteByte)/1024/1024/interval) + " MB/s " +
					" -- " +
					" rb " + readByte +
					" rt " + read_running_threads + 
					" rr " + ((readByte-lastReadByte)/1024/1024/interval) + " MB/s");

				lastWriteByte = writeByte;
				lastReadByte  = readByte;

				try {
					StringBuffer buffer = new StringBuffer();
					buffer.append("<METRIC localtime=\"");
					buffer.append(localtime);
					buffer.append("\" bytesOut=\"");
					buffer.append(lastWriteByte);
					buffer.append("\" bytesIn=\"");
					buffer.append(lastReadByte);
					buffer.append("\"/>\n");
					fileWriter.write(buffer.toString());
				} catch (Exception e) {
					System.out.println(e.toString());
				}
				}
				},
				    0, 
				    interval * 1000);

		/* Start read thread */
		if (readThreads > 0) {
			totalFiles = get_total_filelist();

			for (int i = 0; i < readThreads; ++i) {
				ReadThread tmp = new ReadThread(i);
				readThreadVector.add(tmp);
				tmp.start();
			}
		}

		/* Start write thread */
		if (writeThreads > 0) {
			for (int i = 0; i < writeThreads; ++i) {
				WriteThread tmp = new WriteThread(i);
				writeThreadVector.add(tmp);
				tmp.start();
			}
		}

		/* Waiting all thread exit */
		System.out.println("MAIN: waiting all threds exit....[ write " + writeThreads + " read " + readThreads + " ]");
		try {
			if (writeThreads > 0) {
				for (WriteThread t:writeThreadVector) t.join();
				write_filelist();
			}
			if (readThreads > 0)
				// XXX Why blocked here???
				for (ReadThread t:readThreadVector) t.join();
		} catch (InterruptedException e) {
			System.err.println(e.toString()); 
		}

		System.out.println("MAIN: cancel Timer......");
		schedule.cancel();

		write_io_end();

		// write_stage_environment();

		if (report.equalsIgnoreCase("True")) 
			write_report();

		System.out.println("MAIN: The end!");
	}

	private void ls_dir(Vector<String> result, Path src, boolean recursive) {
		try {
			FileStatus items[] = myDFS.listStatus(src);
			if ((items == null) || ((items.length == 0) && (!myDFS.exists(src)))) {
				return;
			}

			for (int i = 0; i < items.length; i++) {
				FileStatus stat = items[i];
				Path cur = stat.getPath();

				if (recursive) {
					if (!stat.isDir()) {
						result.add(cur.toUri().getPath());
					}
				} else {
					if (stat.isDir()) {
						result.add(cur.toUri().getPath());
					}
				}

				if (recursive && stat.isDir()) {
					ls_dir(result, cur, recursive);
				}
			}
		} catch(IOException e) {
			System.err.println(e.toString());
		}
	}

	/**
	 * From @step @readSource obtain write file list.
	 * @return file list 
	 */
	private Vector<String> get_total_filelist() throws IOException {
		String whichClient = null;
		int i = 0;

		/** Get client list */
		Vector<String> clientList = new Vector<String>();
		ls_dir(clientList, new Path(testCaseNo), false);

		Collections.sort(clientList);

		/** find which client to read */
		for (i = 0; i < clientList.size(); i++) {
			String cli = clientList.get(i);
			if (cli.substring(cli.length()-clientNo.length()).equals(clientNo))
				break;
		}

		if (step < 0) {
			/** All client read one file list */
			step = -step;
			step = step % clientList.size();

			if (step == 0)
				step = clientList.size();
			whichClient = clientList.get(step-1);
		} else {
			/** According step to determine which filelist to read */
			whichClient = clientList.get((i+step) % clientList.size());
		}

		/** get files to read */
		Vector<String> results = new Vector<String>();
		String src = whichClient + "/f" + fileSize;
		ls_dir(results, new Path(src), true);

		System.out.println("READ: " + src + ", there are [ " + results.size() + " ] files");

		// for debug
		write_total_flist(results);

		return results;
	}

	private String build_local_path() {
		String path = shareDir;
		path += "/" + testCaseNo;

		if (globalPrefix != null && !globalPrefix.isEmpty())
			path += "/" + globalPrefix;
		path += "/d" + totalDataNode;
		path += "/c" + totalClient;
		path += "/f" + fileSize;
		if (writeCon == null)
			path += "/con" + writeThreads;
		else
			path += "/con" + readThreads; //wangxu change it from writeCon to readThreads
		path += "/r" + round;
		path += "/"  + clientNo;

		return path;
	}

	private String build_hdfs_path(int i) {
		String path = testCaseNo;
		path += "/"  + clientNo;
		path += "/f" + fileSize;
		if (globalPrefix != null && !globalPrefix.isEmpty())
			path += "/" + globalPrefix;
		path += "/d"   + totalDataNode;
		path += "/c"   + totalClient;
		path += "/con" + writeThreads;
		path += "/r"   + round;
		path += "/stage" + stage + "_" + i;

		return path;
	}

	private String build_read_path() {
		String path = testCaseNo;

		if (globalPrefix != null && !globalPrefix.isEmpty())
			path += "/" + globalPrefix;
		path += "/d" + totalDataNode;
		path += "/c" + totalClient;
		path += "/f" + fileSize;
		path += "/con" + writeCon;
		path += "/r" + round;

		return path;
	}

	/**
	 * Give the stage environment 
	 */
	private void write_stage_environment() throws IOException {
		System.out.println("********* write stage environment ********");

		StringBuffer buffer = new StringBuffer();

		String path = build_local_path();

		File f = new File(path);
		if (!f.exists()) 
			f.mkdirs();

		path += "/"  + stage + ".xml";

		buffer.append("<?xml version=\"1.0\"?>\n");
		buffer.append("<IOTEST>\n");
		buffer.append("<environment>\n");
		buffer.append("<shareDir>" + shareDir + "</shareDir>\n");
		buffer.append("<globalPrefix>" + globalPrefix + "</globalPrefix>\n");
		buffer.append("<testCaseNo>" + testCaseNo + "</testCaseNo>\n");
		buffer.append("<totalDataNode>" + totalDataNode + "</totalDataNode>\n");
		buffer.append("<fileSize>" + fileSize + "</fileSize>\n");
		buffer.append("<expired>" + expired + "</expired>\n");
		buffer.append("<readThreads>" + readThreads + "</readThreads>\n");
		buffer.append("<writeThreads>" + writeThreads+ "</writeThreads>\n");
		buffer.append("<round>" + round + "</round>\n");
		buffer.append("<clientNo>" + clientNo + "</clientNo>\n");
		buffer.append("<report>" + report + "</report>\n");
		buffer.append("<readMode>" + readMode + "</readMode>\n");
		buffer.append("<writeCon>" + writeCon + "</writeCon>\n");
		buffer.append("<step>" + step + "</step>\n");
		buffer.append("<stage>" + stage + "</stage>\n");
		buffer.append("<readSource>" + readSource + "</readSource>\n");
		buffer.append("</environment>\n");
		buffer.append("</IOTEST>\n");

		write_to_file(path, buffer.toString());
	}


	/**
	 * Gives a report on how the FileSystem is doing after finish a stage.
	 * @exception IOException if the filesystem does not exits.
	 */
	private void write_report() throws IOException {
		System.out.println("****************** report *******************");
		StringBuffer buffer = new StringBuffer();
		String path = build_local_path();

		File f = new File(path);
		if (!f.exists()) 
			f.mkdirs();

		path += "/" + "stage_" + stage + "_report.xml";

		DiskStatus ds = myDFS.getDiskStatus();

		long raw = ds.getCapacity();
		long rawUsed = ds.getDfsUsed();
		long remaining = ds.getRemaining();
		long used = myDFS.getUsed();
		boolean mode = myDFS.setSafeMode(FSConstants.SafeModeAction.SAFEMODE_GET);

		UpgradeStatusReport status = myDFS.distributedUpgradeProgress(UpgradeAction.GET_STATUS);

		buffer.append("<?xml version=\"1.0\"?>\n");
		buffer.append("<report>\n");
		buffer.append("<nameNode>\n");

		if (mode) {
			buffer.append("<safeMode>on</safeMode>\n");
		} else {
			buffer.append("<safeMode>off</safeMode>\n");
		}

		if (status != null) {
			buffer.append("<status>" + 
					status.getStatusText(false) +
					"</status>\n");
		}

		buffer.append("<totalRawByte>" + raw + "</totalRawByte>\n");
		buffer.append("<remainRawByte>" + remaining + "</remainRawByte>\n");
		buffer.append("<usedRawByte>" + rawUsed + "</usedRawByte>\n");
		buffer.append("<totalEffectiveByte>" + used + 
				"</totalEffectiveByte>\n");

		buffer.append("</nameNode>\n");

		DatanodeInfo[] info = myDFS.getDataNodeStats();
		buffer.append("<cluster available=\"" + info.length + "\">\n");      
		for (int i = 0; i < info.length; i++) {
			buffer.append("<datanode name=\"" + info[i].getHostName() +
					"\" rackName=\"" + info[i].getNetworkLocation() +
					"\" totalRawByte=\"" + info[i].getCapacity() +
					"\" remainRawByte=\"" + info[i].getRemaining() +
					"\" usedByte=\"" + info[i].getDfsUsed() +
					"\"/>\n");
		}

		buffer.append("</cluster>\n");
		buffer.append("</report>\n");

		String report_xml = buffer.toString();
		//        System.out.println(report_xml);

		write_to_file(path, report_xml);
	}

	private void write_to_file(String path, String content) throws IOException {
		FileWriter f = null;
		try{
			f = new FileWriter(path);
			f.write(content, 0, content.length());
		} catch (IOException e) {
			System.err.print(e.toString());
		} finally {
			if (null != f) f.close();
		}
	}

	private void write_io_begin() throws IOException {
		String path = build_local_path();

		File f = new File(path);
		if (!f.exists()) 
			f.mkdirs();

		path += "/stage_" + stage + ".xml";
		System.out.println("xml path: " + path);

		if (null == fileWriter)
			fileWriter = new FileWriter(path);
		fileWriter.write("<?xml version=\"1.0\"?>\n");
		fileWriter.write("<IOTEST>\n");
	}

	private void write_io_end() throws IOException {
		if (null != fileWriter) {
			fileWriter.write("</IOTEST>\n");
			fileWriter.flush();
			fileWriter.close();
		}
	}

	static void printUsage () {
		System.out.println("Usage : IOtask <read thread(int)> <write thread>(int)");
		System.exit(2);
	}

	public static void main(String[] argv) {

		String sdir = null;
		String tno = null;
		String prefix = null;
		int dnum = 0;
		int cnum = 0;
		int rno = 0;
		String cno = null;
		int expire = 20;
		String report = "False";
		String stage = null;

		int wthread = 0;
		int fsize = 1024;

		String rsrc = null;
		String wcon = null;
		int rthread = 0;
		int rmode = 0;
		int step = 0;
		long totalbyte = 0;

		String[] tmp = null;

		for (int i = 0; i < argv.length; ++i) {
			if (argv[i].equalsIgnoreCase("-sdir"))
				sdir = argv[++i];
			else if (argv[i].equalsIgnoreCase("-tno"))
				tno = argv[++i];
			else if (argv[i].equalsIgnoreCase("-prefix"))
				prefix = argv[++i];
			else if (argv[i].equalsIgnoreCase("-dnum"))
				dnum = Integer.parseInt(argv[++i]);
			else if (argv[i].equalsIgnoreCase("-cnum"))
				cnum = Integer.parseInt(argv[++i]);
			else if (argv[i].equalsIgnoreCase("-rno"))
				rno = Integer.parseInt(argv[++i]);
			else if (argv[i].equalsIgnoreCase("-cno"))
				cno = argv[++i];
			else if (argv[i].equalsIgnoreCase("-expire"))
				expire = Integer.parseInt(argv[++i]);
			else if (argv[i].equalsIgnoreCase("-report"))
				report = argv[++i];
			else if (argv[i].equalsIgnoreCase("-w"))
				wthread = Integer.parseInt(argv[++i]);
			else if (argv[i].equalsIgnoreCase("-fsize"))
				fsize = Integer.parseInt(argv[++i]);
			else if (argv[i].equalsIgnoreCase("-r"))
				rthread = Integer.parseInt(argv[++i]);
			else if (argv[i].equalsIgnoreCase("-stage"))
				stage = argv[++i];
			else if (argv[i].equalsIgnoreCase("-rsrc"))
				rsrc = argv[++i];
			else if (argv[i].equalsIgnoreCase("-rmode"))
				rmode = Integer.parseInt(argv[++i]);
			else if (argv[i].equalsIgnoreCase("-step"))
				step = Integer.parseInt(argv[++i]);
			else if (argv[i].equalsIgnoreCase("-totalbyte"))
				totalbyte = Long.parseLong(argv[++i]);
			else
				System.err.println(argv[i]);
		}

		if (rthread > 0 && rsrc != "null") {
			tmp = rsrc.split(",");
			try {
				wcon = tmp[0];
				rsrc = tmp[1];
			} catch (ArrayIndexOutOfBoundsException e) {
				System.out.println(e.toString());
				printUsage();
				System.exit(1);
			}
		}

		if (rthread > 0 && (wcon == null || rsrc == "null")) {
			printUsage();
			System.exit(1);
		}

		// verify options
		if (sdir == null || sdir.isEmpty() ||
				tno == null || tno.isEmpty() || 
				cno == null || cno.isEmpty() ||
				dnum <= 0 || cnum <= 0 || fsize <= 0 || expire <= 0 ||
				rno < 0 ) {
			printUsage();
			System.exit(1);
		}

		try {
			TestIO mytask = new TestIO(sdir, prefix, tno, dnum, cnum, 
					fsize, expire, rthread, wthread, rno, cno, report, 
					rmode, step, wcon, stage, rsrc, totalbyte);

			mytask.start();
		} catch (Exception e) {
			System.err.println(e.toString());
		}
	}
}
